package com.atbeijing.D08;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternFlatSelectFunction;
import org.apache.flink.cep.PatternFlatTimeoutFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.List;
import java.util.Map;

/**
 * 使用CEP实现订单超时需求
 */
public class Example8 {
    //测输出流,也叫旁路输出
    private static OutputTag<String> stringOutputTag = new OutputTag<String>("timeout-order"){};

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //数据流
        SingleOutputStreamOperator<Event> stream = env
                .fromElements(
                        new Event("order-1", "create", 1000L),
                        new Event("order-2", "create", 2000L),
                        new Event("order-1", "pay", 4000L)
                )
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                                    @Override
                                    public long extractTimestamp(Event element, long recordTimestamp) {
                                        return element.timestamp;
                                    }
                                })
                );

        //定义模板
        Pattern<Event, Event> pattern = Pattern
                .<Event>begin("create")//第一个事件取名字
                .where(new SimpleCondition<Event>() {//要求
                    @Override
                    public boolean filter(Event value) throws Exception {
                        return value.eventType.equals("create");//符合要求的时间写入list
                    }
                })
                .next("pay")//紧接着的下个事件取名
                .where(new SimpleCondition<Event>() {//要求
                    @Override
                    public boolean filter(Event value) throws Exception {
                        return value.eventType.equals("pay");//符合要求的时间写入list
                    }
                })
                .within(Time.seconds(5));//要求两个事件紧接着,且在5秒内发生

        //先分流,为流匹配模板
        PatternStream<Event> patternStream = CEP.pattern(stream.keyBy(r -> r.orderId), pattern);

        //通过模板处理后的数据
        SingleOutputStreamOperator<String> result = patternStream.flatSelect(
                stringOutputTag,//超时信息输出到测输出标签
                //处理超时数据
                new PatternFlatTimeoutFunction<Event, String>() {
                    @Override
                    public void timeout(Map<String, List<Event>> pattern, long timeoutTimestamp, Collector<String> out) throws Exception {
                        Event create = pattern.get("create").get(0);
                        //将数据发送到侧输出流
                        out.collect("订单" + create.orderId + "支付超时!!");
                    }
                },
                // 处理正常的事件
                new PatternFlatSelectFunction<Event, String>() {
                    @Override
                    public void flatSelect(Map<String, List<Event>> pattern, Collector<String> out) throws Exception {
                        Event pay = pattern.get("pay").get(0);
                        //将数据向下游发送
                        out.collect("订单" + pay.orderId + "支付成功!!!");
                    }
                }
        );

        //打印侧输出流,处理失败的数据
        result.getSideOutput(stringOutputTag).print();

        //主流打印支付成功数据
        result.print();

        env.execute();

    }

    public static class Event {
        public String orderId;
        public String eventType;
        public Long timestamp;

        public Event() {
        }

        public Event(String orderId, String eventType, Long timestamp) {
            this.orderId = orderId;
            this.eventType = eventType;
            this.timestamp = timestamp;
        }

        @Override
        public String toString() {
            return "Event{" +
                    "orderId='" + orderId + '\'' +
                    ", eventType='" + eventType + '\'' +
                    ", timestamp=" + timestamp +
                    '}';
        }
    }
}
